use paho_mqtt::{CreateOptionsBuilder, AsyncClient, ServerResponse, QOS_2, Message};
use std::time::Duration;
use tokio::task;
use futures_util::StreamExt;
use std::sync::mpsc;
use std::sync::mpsc::{Receiver};

/// 创建异步mqtt客户端
pub fn create_client(server_host: &str, server_port: u16, client_id: &str) -> paho_mqtt::Result<AsyncClient> {
    CreateOptionsBuilder::new()
        .server_uri(format!("tcp://{}:{}", server_host, server_port))
        .client_id(client_id)
        .mqtt_version(5)
        .create_client()
}

/// 链接异步客户端
pub async fn connect(client: &AsyncClient) -> paho_mqtt::Result<ServerResponse> {
    let conn_opts = paho_mqtt::ConnectOptionsBuilder::new()
        .keep_alive_interval(Duration::from_secs(60))
        .clean_session(false)
        .automatic_reconnect(
            Duration::from_secs(10),
            Duration::from_secs(30),
        )
        .finalize();
    client.connect(conn_opts).await
}

/// 订阅主题并监听
pub async fn subscribe<T>(client: &mut AsyncClient, topic: T) -> paho_mqtt::Result<Receiver<Message>>
    where T: Into<String>
{
    client.subscribe(topic, QOS_2).await?;
    let mut receiver = client.get_stream(64);
    let (s, r) = mpsc::channel();
    task::spawn(async move {
        loop {
            match receiver.next().await {
                Some(Some(msg)) => {
                    if let Err(e) = s.send(msg) {
                        println!("{}", e);
                        break;
                    }
                }
                _ => { continue; }
            }
        }
    });
    Ok(r)
}

/// 发送消息
pub async fn publish<T, P>(client: &mut AsyncClient, topic: T, payload: P) -> paho_mqtt::Result<()>
    where T: Into<String>,
          P: Into<Vec<u8>>
{
    let message = Message::new(topic, payload, QOS_2);
    client.publish(message).await
}

#[cfg(test)]
mod tests {
    use paho_mqtt::ReasonCode;
    use crate::mqtt::basic;

    #[tokio::test]
    async fn test() {
        let mut client = basic::create_client("42.193.40.147", 1883, "squb").unwrap();
        let resp = basic::connect(&client).await.unwrap();
        assert!(resp.reason_code().eq(&ReasonCode::Success));
        let receiver = basic::subscribe(&mut client, "test").await.unwrap();
        loop {
            match receiver.recv() {
                Ok(msg) => {
                    println!("{}", msg);
                }
                Err(e) => {
                    println!("{}", e);
                }
            }
        }
    }
}